package com.hngy.scala.stream

import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.streaming.api.windowing.time.Time


/**
  * 需求：通过Socket实时产生一些单词，
  * 使用Flink实时接收数据
  * 对指定时间窗口内(例如：2秒)的数据进行聚合统计
  * 并且把时间窗口内计算的结果打印出来
  */
object SocketWindowWordCountScala {

  //注意：在执行代码之前，需要先在hadoop001机器上开启socket，端口为9002
  def main(args: Array[String]): Unit = {
    //获取运行环境
    val env = StreamExecutionEnvironment.getExecutionEnvironment

    //连接socket获取输入数据
    val text = env.socketTextStream("hadoop001",9002)

    //处理数据
    //注意：必须要添加这一行隐式转换的代码，否则下面的flatMap方法会报错
    import org.apache.flink.api.scala._
    val wordCount = text.flatMap(_.split(" ")) //将每一行有空格分隔成单词
      .map((_,1))
      .keyBy(tup=>tup._1)
      .timeWindow(Time.seconds(2)) //时间窗口为2秒，表示每隔2秒钟计算一次接收到的数据
      .sum(1)// 使用sum或者reduce都可以

    //使用一个线程执行打印操作
    wordCount.print().setParallelism(1)

    //执行程序
    env.execute("SocketWindowWordCountScala")
  }
}
